package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class TestApp extends BaseSqlApp {
    public static void main(String[] args) {
        new TestApp().init(10033, 3, "TestApp");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
//        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(20));
        tableEnv.executeSql(
                "create table t1 (\n" +
                "    `id` string,\n" +
                "    `name` string,\n" +
                "    `code` string,\n" +
                "    `pt` as proctime()\n" +
                ") with (\n" +
                "    'connector' = 'kafka',\n" +
                "    'properties.bootstrap.servers' = 'hadoop102:9092,hadoop103:9092,hadoop104:9092',\n" +
                "    'properties.group.id' = 'tsing',\n" +
                "    'topic' = 's1',\n" +
                "    'format' = 'csv'\n" +
                ")"
        );

        tableEnv.executeSql(
                "create table t2 (\n" +
                "    `id` string,\n" +
                "    `age` string\n" +
                ") with (\n" +
                "    'connector' = 'kafka',\n" +
                "    'properties.bootstrap.servers' = 'hadoop102:9092,hadoop103:9092,hadoop104:9092',\n" +
                "    'properties.group.id' = 'tsing',\n" +
                "    'topic' = 's2',\n" +
                "    'format' = 'csv'\n" +
                ")"
        );

        readBaseDicFromMysql(tableEnv);

        tableEnv.sqlQuery(
                "select t1.id, name, code, age, dic_name " +
                "from t1 " +
                "join t2 on t1.id = t2.id " +
                "join base_dic for system_time as of t1.pt as bd on t1.code = bd.dic_code"
        ).execute().print();
    }
}
